并发编程

并发引入

并发的需要

并发的问题

基本的并发方法

Process-based

并发编程

优劣分析

1. Pros

   • Handle multiple connections concurrently

   • Clean sharing model

   • Simple and straightforward

2. Cons

   • Additional overhead for process control

   • Nontrivial to share data between processes（IPC）

并发服务器

1. 基本思想

   

2. 代码实现

   xxxxxxxxxx
   35
   1
   #include "csapp.h"
   2
   void echo(int connfd);
   3
   4
   void sigchld_handler(int sig)
   5
   {
   6
       while (waitpid(-1, 0, WNOHANG) > 0)
   7
           ; // reap all zombie children
   8
       return;
   9
   }
   10
   11
   int main(int argc, char **argv)
   12
   {
   13
       int listenfd, connfd;
   14
       socklen_t clientlen;
   15
       struct sockaddr_storage clientaddr;
   16
   17
       if (argc != 2) {
   18
           fprintf(stderr, "usage: %s <port>\n", argv[0]);
   19
           exit(0);
   20
       }
   21
   22
       Signal(SIGCHLD, sigchld_handler);
   23
       listenfd = Open_listenfd(argv[1]);
   24
       while (1) {
   25
           clientlen = sizeof(struct sockaddr_storage);
   26
           connfd = Accept(listenfd, (SA *) &clientaddr, &clientlen);
   27
           if (Fork() == 0) {
   28
               Close(listenfd); /* Child closes its listening socket */
   29
               echo(connfd);    /* Child services client */
   30
               Close(connfd);   /* Child closes connection with client */
   31
               exit(0);         /* Child exits */
   32
           }
   33
           Close(connfd); /* Parent closes connected socket (important!) */
   34
       }
   35
   }
   

Event-based

并发编程

实现对于fd数组的维护+select调用

xxxxxxxxxx
7
1
int select(int n, fd_set *fdset, NULL, NULL, NULL);
2
// 返回已准备好的描述符的非零的个数，若出错则为 -1。
3
4
FD_ZERO(fd_set *fdset);           /* Clear all bits in fdset */
5
FD_CLR(int fd, fd_set *fdset);    /* Clear bit fd in fdset */
6
FD_SET(int fd, fd_set *fdset);    /* Turn on bit fd in fdset */
7
FD_ISSET(int fd, fd_set *fdset);  /* Is bit fd in fdset on? */

• nfds：等待的最大文件描述符 +1

• readfds：不为 NULL 时检测集合内可读文件

• writefds：不为 NULL 时检测集合内可写文件

• exceptfds：不为 NULL 时检测集合内异常文件timeout：若不为 NULL，用来设置阻塞的最大时间返回值大于 0 表示规定时间内有文件被检测，等于 0 表示超时，小于 0 表示发生错误

并发服务器

xxxxxxxxxx
41
1
#include "csapp.h"
2
void echo(int connfd);
3
void command(void);
4
5
int main(int argc, char **argv)
6
{
7
    int listenfd, connfd;
8
    socklen_t clientlen;
9
    struct sockaddr_storage clientaddr;
10
    fd_set read_set, ready_set;
11
    
12
    if (argc != 2) {
13
        fprintf(stderr, "usage: %s <port>\n", argv[0]);
14
        exit(0);
15
    }
16
    listenfd = Open_listenfd(argv[1]);
17
    
18
    FD_ZERO(&read_set);              /* Clear read set */
19
    FD_SET(STDIN_FILENO, &read_set); /* Add stdin to read set */
20
    FD_SET(listenfd, &read_set);     /* Add listenfd to read set */
21
22
    while (1) {
23
        ready_set = read_set;
24
        Select(listenfd + 1, &ready_set, NULL, NULL, NULL);
25
        if (FD_ISSET(STDIN_FILENO, &ready_set))
26
            command(); /* Read command line from stdin */
27
        if (FD_ISSET(listenfd, &ready_set)) {
28
            clientlen = sizeof(struct sockaddr_storage);
29
            connfd = Accept(listenfd, (SA *)&clientaddr, &clientlen);
30
            echo(connfd); /* Echo client input until EOF */
31
            Close(connfd);
32
        }
33
    }
34
}
35
36
void command(void) {
37
    char buf[MAXLINE];
38
    if (!Fgets(buf, MAXLINE, stdin))
39
        exit(0); /* EOF */
40
    printf("%s", buf); /* Process the input command */
41
}

Thread-based

并发编程

• printf是线程安全函数——进程调度的基本单位是线程，单锁不再会导致死锁

创建线程

xxxxxxxxxx
2
1
typedef void *(func)(void *);
2
int pthread_create(pthread_t *tid, pthread_attr_t *attr, func *f, void *arg);

• ``参数

终止线程

xxxxxxxxxx
4
1
void pthread_exit(void *thread_return);
2
3
pthead_t pthread_self(void); //终止自身进程，主进程会等待所有其他对等线程终止
4
int pthread_cancel(pthread_t tid); //终止tid线程

• ``参数

回收线程

xxxxxxxxxx
1
1
int pthread_join(pthread_t tid, void **thread_return);

• ``参数

分离线程

xxxxxxxxxx
1
1
int pthread_detach(pthread_t tid);

• ``参数

初始化线程

xxxxxxxxxx
2
1
pthread_once_t once_control = PTHREAD_ONCE_INIT;
2
int pthread_once(pthread_once_t *once_control, void (*init_routine)(void));

• ``参数

并发服务器

共享变量

无意识共享问题

信号量同步

循环体拆分

进度图

信号量

调度共享资源

生产者-消费者问题

xxxxxxxxxx
9
1
typedef struct {
2
    int *buf;       /* Buffer array */
3
    int n;          /* Maximum number of slots */
4
    int front;      /* buf[(front+1)%n] is first item */
5
    int rear;       /* buf[rear%n] is last item */
6
    sem_t mutex;    /* Protects accesses to buf */
7
    sem_t slots;    /* Counts available slots */
8
    sem_t items;    /* Counts available items */
9
} sbuf_t;

xxxxxxxxxx
41
1
#include "csapp.h"
2
#include "sbuf.h"
3
4
/* Create an empty, bounded, shared FIFO buffer with n slots */
5
void sbuf_init(sbuf_t *sp, int n)
6
{
7
    sp->buf = Calloc(n, sizeof(int));
8
    sp->n = n;                       /* Buffer holds max of n items */
9
    sp->front = sp->rear = 0;        /* Empty buffer iff front == rear */
10
    Sem_init(&sp->mutex, 0, 1);      /* Binary semaphore for locking */
11
    Sem_init(&sp->slots, 0, n);      /* Initially, buf has n empty slots */
12
    Sem_init(&sp->items, 0, 0);      /* Initially, buf has zero data items */
13
}
14
15
/* Clean up buffer sp */
16
void sbuf_deinit(sbuf_t *sp)
17
{
18
    Free(sp->buf);
19
}
20
21
/* Insert item onto the rear of shared buffer sp */
22
void sbuf_insert(sbuf_t *sp, int item)
23
{
24
    P(&sp->slots);                           /* Wait for available slot */
25
    P(&sp->mutex);                           /* Lock the buffer */
26
    sp->buf[(++sp->rear) % (sp->n)] = item;  /* Insert the item */
27
    V(&sp->mutex);                           /* Unlock the buffer */
28
    V(&sp->items);                           /* Announce available item */
29
}
30
31
/* Remove and return the first item from buffer sp */
32
int sbuf_remove(sbuf_t *sp)
33
{
34
    int item;
35
    P(&sp->items);                           /* Wait for available item */
36
    P(&sp->mutex);                           /* Lock the buffer */
37
    item = sp->buf[(++sp->front) % (sp->n)]; /* Remove the item */
38
    V(&sp->mutex);                           /* Unlock the buffer */
39
    V(&sp->slots);                           /* Announce available slot */
40
    return item;
41
}

读者-写者问题

xxxxxxxxxx
35
1
/* Global variables */
2
int readcnt;    /* Initially = 0 */
3
sem_t mutex, w; /* Both initially = 1 */
4
5
void reader(void)
6
{
7
    while (1) {
8
        P(&mutex);
9
        readcnt++;
10
        if (readcnt == 1) /* First in */
11
            P(&w);
12
        V(&mutex);
13
14
        /* Critical section */
15
        /* Reading happens  */
16
17
        P(&mutex);
18
        readcnt--;
19
        if (readcnt == 0) /* Last out */
20
            V(&w);
21
        V(&mutex);
22
    }
23
}
24
25
void writer(void)
26
{
27
    while (1) {
28
        P(&w);
29
        
30
        /* Critical section */
31
        /* Writing happens  */
32
33
        V(&w);
34
    }
35
}

其它并发问题

性能刻画

• 强扩展，弱扩展

• 加速比：绝对加速比(T1顺序)+相对加速比(T1并行)

  $$S_p=\frac{T_1}{T_p}$$

• 效率

  $$E_p=\frac{S_p}{p}=\frac{T_1}{p{T}_k}$$

线程安全

1. 不保护共享变量的的函数

2. 保持跨越多个调用的状态的函数

3. 返回指向静态变量的指针的函数

4. 调用线程不安全函数的函数

可重入性

• 显示可重入的：传值传递，所有数据引用都是本地自动栈变量

• 饮食可重入的：允许传递指针

竞争

• 当一个程序的正确性依赖于一个线程要在另一个线程到达y点之前到达它的控制流中的x点时，就会发生竞争

死锁

• 例如：每个线程都先对S加锁，再对T加锁